/*
 * NetService.cpp
 *
 *  Created on: Jan 21, 2013
 *      Author: root
 */

#include "NetService.h"
#define CORE_VER "1.0.30" /*core version*/

/**
 * class init
 */
NetService::NetService()
{
		svr_run_flag = 0;
		limit_thread_count = LIMIT_MAX_THREADS;
		max_connect_num = DFLT_MIN_CONN_SIZ;

		listenfd = 0;
		listenport = 8080;
		epfd = -1;
		workThreadId = 0;
		n_event_thread_num = 0;
		n_send_thread_num = 0;
		memset(eventThreadId,NULL,sizeof(eventThreadId));
		memset(sendThreadId,NULL,sizeof(sendThreadId));
		memset(listenIP,'\0',sizeof(listenIP));

		/*Semaphore  init*/
		sem_init(&io_control_sem, 0, 0);
		sem_init(&io_event_sem, 0, 0);
		sem_init(&io_send_sem, 0, 0);

}

/**
 * class free
 */
NetService::~NetService()
{
		this->log_fmt(LOG_INFO,"NetService exit! \r\n");
		/**
		 * server stop flag
		 */
		svr_run_flag = 0;

		/*Semaphore  destroy*/
		sem_destroy(&io_control_sem);
		sem_destroy(&io_event_sem);
		sem_destroy(&io_send_sem);

		/**
		 * close listen
		 */
		if(listenfd != 0)
		{
			close(listenfd);
			listenfd = NULL;
		}
		/**
		 * close epool fd
		 */
		if(epfd >= 0)
		{
			close(epfd);
			epfd = -1;
		}
}

/*
 *
 * init net service args
 */
int NetService::init(const char * listenip,uint16_t port,int max_connect,int max_threads_count)
{
		if(listenip == NULL)
				return FAILURE;

		strcpy(this->listenIP,listenip);
		this->listenport = port;
		this->max_connect_num = max_connect;
		this->limit_thread_count = max_threads_count;
		if(limit_thread_count > LIMIT_MAX_THREADS)
		{
			limit_thread_count = LIMIT_MAX_THREADS;
		}
		/*
		 * init svr rc
		 */
		if(this->init_epool()  != SUCCESS)
		{
				this->log_fmt(LOG_ERR,"init_epool,FAILURE!");
				return FAILURE;
		}

		if(this->init_listen() != SUCCESS)
		{
				this->log_fmt(LOG_ERR,"init_listen,FAILURE!");
				return FAILURE;
		}

		if(connpool.init_conn_pool(max_connect) != SUCCESS)
		{
				this->log_fmt(LOG_ERR,"init_conn_pool,FAILURE!");
				return FAILURE;
		}

		if(evtqueue.init_queue() != SUCCESS)
		{
				this->log_fmt(LOG_ERR,"evtqueue init_queue,FAILURE!");
				return FAILURE;
		}

		if(net_op_queue.init_queue() != SUCCESS)
		{
				this->log_fmt(LOG_ERR,"net_op_queue init_queue,FAILURE!");
				return FAILURE;
		}

		if(io_queue.init_queue() != SUCCESS)
		{
				this->log_fmt(LOG_ERR,"io_queue init_queue,FAILURE!");
				return FAILURE;
		}

		this->log_fmt(LOG_INFO,"net service init ok , epoll fd:%d,listen fd:%d ,listen port:%d\r\n",epfd,listenfd,port);

		return SUCCESS;
}

/*
 *
 * start net service
 */
int NetService::start()
{
		svr_run_flag = 1;
		pthread_attr_t attr;
		pthread_attr_init(&attr);
		pthread_attr_setscope(&attr, PTHREAD_SCOPE_SYSTEM);
		pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
		int ret = 0;
		/*start work thread*/
		ret = pthread_create(&workThreadId, &attr,io_working_proces,this);
		if(ret != 0)
		{
			close(listenfd);
			close(epfd);
			return FAILURE;
		}
		else
		{
			this->log_fmt(LOG_INFO,"work thread id:%u \r\n",(unsigned) workThreadId);
		}

		/*start event thread*/
		for(int i = 0;i < this->limit_thread_count  &&  i < LIMIT_MAX_THREADS; i++)
		{
			ret = pthread_create(this->eventThreadId + n_event_thread_num , &attr , io_event_proces,this);

			if(ret != 0)
			{
			   close(listenfd);
			   close(epfd);
			   return FAILURE;
			}

			this->log_fmt(LOG_INFO,"event thread id:%u \r\n",(unsigned) this->eventThreadId[i]);

			n_event_thread_num ++;
		}

		/*start clean thread*/
		ret = pthread_create(&clearThreadId, &attr , io_clean_proces , this);
		if(ret != 0)
		{
			close(listenfd);
			close(epfd);
			return FAILURE;
		}
		else
		{
			this->log_fmt(LOG_INFO,"clean thread id:%u \r\n",(unsigned) clearThreadId);
		}


		/*start net operation thread*/
		ret = pthread_create(&netoperationThreadId, &attr , net_opration_proces , this);
		if(ret != 0)
		{
		   close(listenfd);
		   close(epfd);
		   return FAILURE;
		}
		else
		{
		   this->log_fmt(LOG_INFO,"netopration thread id:%u \r\n",(unsigned) netoperationThreadId);
		}

		/*start io control thread*/
		ret = pthread_create(&controlThreadId, &attr , io_control_proces , this);
		if(ret != 0)
		{
		   close(listenfd);
		   close(epfd);
		   return FAILURE;
		}
		else
		{
		   this->log_fmt(LOG_INFO,"control thread id:%u \r\n",(unsigned) controlThreadId);
		}

		/*start send thread*/
		for(int i = 0;i < this->limit_thread_count && i < LIMIT_MAX_THREADS ;i++)
		{
			 ret = pthread_create(this->sendThreadId + n_send_thread_num , &attr , io_sendfile_proces,this);

			 if(ret != 0)
			 {
				close(listenfd);
				close(epfd);
				return FAILURE;
			 }

			 this->log_fmt(LOG_INFO,"send thread id:%u \r\n",(unsigned) this->sendThreadId[i]);

			 n_send_thread_num ++;
		}

		/*
		 * return start listen
		 */
		return start_listen();
}

/*
 *
 * stop net service
 */
int NetService::stop()
{
		svr_run_flag = 0;
		usleep(THREAD_TIME_TICK);

		/*close listen*/
		close_listen();

		/*close all thread*/
		for(int i=0 ;i < this->n_event_thread_num;i++)
		  sem_post(&this->io_event_sem);

		for(int i=0 ;i < this->n_send_thread_num;i++)
			 sem_post(&this->io_send_sem);

		sem_post(&this->io_control_sem);

		/*wait 1 seconds*/
		usleep(1000000);
		return SUCCESS;
}

/*
*inherit virtual function
*/
NEXT_OPRATION NetService::accept_completed(user_token *& token,buf_chunk * sock_buf)
{
    return DO_RECV;
}
/*
*connect_completed
*/
NEXT_OPRATION NetService::connect_completed(user_token * token,buf_chunk * sock_buf)
{
     return DO_SEND;
}
/*
*receive_completed
*/
NEXT_OPRATION  NetService::receive_completed(user_token * token,buf_chunk * sock_buf)
{
    return DO_CLOSE;
}
/*
*send_completed
*/
NEXT_OPRATION NetService::send_completed(user_token * token,buf_chunk * sock_buf)
{
    return DO_CLOSE;
}
/**
 *trans_heartbeat
 */
NEXT_OPRATION NetService::trans_heartbeat(user_token * token)
{
	return DO_TRANSFILE;
}
/*
*trans_completed
*/
NEXT_OPRATION  NetService::trans_completed(user_token * token,buf_chunk * sock_buf)
{
    return DO_CLOSE;
}
/*
*send_wait_compledted
*/
NEXT_OPRATION NetService::recv_or_send_wait_completed(user_token * token,buf_chunk * sock_buf)
{
    return DO_WAIT;
}
/*
*send_wait_compledted
*/
void NetService::close_completed(user_token * token,buf_chunk * sock_buf,NET_ERROR_CODE error)
{
    /*add your code here*/
}
/*
*svr log
*/
int  NetService::log_fmt(int nloglvl,const char * lpszFormat , ...)
{
    return SUCCESS;
}


/*
*svr function
*/
int NetService::get_connect_pool_count()
{
     return this->connpool.get_total_socket_size();
}
/*
*svr function
*/
int NetService::get_connect_pool_use_count()
{
     return this->connpool.conn_list.size() + this->connpool.wait_list.size();
}
/*
*svr function
*/
int NetService::get_connect_pool_unuse_count()
{
     return this->connpool.get_free_socket_size();
}
/*
*svr function
*/
int NetService::get_buf_pool_count()
{
     return this->connpool.mempool.get_count();
}
/*
*svr function
*/
int NetService::get_buf_pool_unuse_count()
{
    return this->connpool.mempool.get_free_count();
}

int NetService::get_current_listen_port()
{
    return this->listenport;
}

const char * NetService::get_core_version()
{
	return CORE_VER;
}

/*
 * set socket noblock
 */
int NetService::setnonblocking(int sock)
{
		int opts;
		opts=fcntl(sock,F_GETFL);
		if(opts < 0)
		{
			return FAILURE;
		}
		opts = opts|O_NONBLOCK;
		if(fcntl(sock,F_SETFL,opts)<0)
		{
			return FAILURE;
		}

		return SUCCESS;
}
/*
 * set socket cork
 */
void NetService::sock_set_cork(int fd)
{
		int on = 1;
		if(-1 == setsockopt(fd, IPPROTO_TCP, TCP_CORK, (char *)&on, sizeof(on)))
		{
			this->log_fmt(LOG_WARN,"warning! sock_set_cork(%s)\n", strerror(errno));
		}
}


/*
 * post user connection
 **/
int NetService::post_usr_connect(user_token * token,NETPROTOCL net_protocl)
{
		if(token == NULL)
		  return FAILURE;

		return net_op_queue.op_enqueue(net_protocl,token);
}

/**
 * init epool fd
 */
int NetService::init_epool()
{
	 	 /*
	     * create epoll mode
	     */
	    if(epfd < 0)
	    {
	        	epfd = epoll_create(MAXEVENTS);
	    }

	    if (epfd < 0)
	    {
				this->log_fmt(LOG_ERR,"epoll_create,FAILURE!");
				return FAILURE;
	    }

	    return SUCCESS;
}
/*
 *
 * init listen socket
 */
int NetService::init_listen()
{
    /*define listen addr */
    struct sockaddr_in listen_addr;
    bzero(&listen_addr,sizeof(struct sockaddr_in));
    listen_addr.sin_family = AF_INET;
    if(strlen(this->listenIP) == 0)
        listen_addr.sin_addr.s_addr = htonl(INADDR_ANY);/*inet_addr(svr_name);*/
    else
        listen_addr.sin_addr.s_addr = inet_addr(this->listenIP);
    listen_addr.sin_port = htons(this->listenport);

    /*
     * create listen socket
     */
    listenfd = socket(AF_INET, SOCK_STREAM, 0);
    if (listenfd <= 0)
    {
		this->log_fmt(LOG_ERR,"create listen socket,FAILURE!");
		return FAILURE;
    }

    /*set socket option*/
    int reuseaddr = 1;
    setsockopt(listenfd,SOL_SOCKET,SO_REUSEADDR,&reuseaddr,sizeof(reuseaddr));

    /*set noblocking*/
    setnonblocking(listenfd);

    /*
     * bind socket
     */
    bool bind_flag = false;
    for(int i = 0; i< 5 ; i++)
    {
    	if (bind(listenfd, (struct sockaddr *)&listen_addr, sizeof(struct sockaddr)) == 0)
			{
					this->log_fmt(LOG_INFO,"bind listen socket,success listen port:%d!",this->listenport);
					bind_flag = true;
					break;
			}
			else
			{
					this->log_fmt(LOG_ERR,"bind listen socket,error:%d!",errno);
					listen_addr.sin_port = htons(this->listenport ++);
			}
    }
    if(!bind_flag)
    {
			close(listenfd);
			listenfd = 0;
			return FAILURE;
    }

    /*bind listen to epoll event*/
	ev.data.fd = listenfd;
	ev.events = EPOLLIN|EPOLLET;
	if(epoll_ctl(epfd, EPOLL_CTL_ADD, listenfd, &ev)  !=  0)
	{
			this->log_fmt(LOG_ERR,"listen epoll_ctl,FAILURE!");
			close(listenfd);
			listenfd = 0;
			return FAILURE;
	}

	/*init end*/
    return SUCCESS;
}
/*
 * start listen
 */
int NetService::start_listen()
{
    /*start listen*/
     int ret = listen(listenfd, 5);
     if (ret != 0)
     {
			 this->log_fmt(LOG_ERR,"listen start,FAILURE!");
			 close(listenfd);
			 listenfd = 0;
			 return FAILURE;
     }

     return SUCCESS;
}
/*
 * close listen ,not close epoll
 */
void NetService::close_listen()
{
		epoll_ctl(epfd, EPOLL_CTL_DEL, listenfd, &ev);
		close(listenfd);
		listenfd = 0;
}
/*
 * re listen,first close listen ,then create new listen,and start listen.
 */
int NetService::re_listen()
{
		int ret = FAILURE;
		if(this->svr_run_flag == 1)
		{
				close_listen();

				if(init_listen() == SUCCESS)
				{
					ret = start_listen();
				}
		}
		return ret;
}


/*
 * post TCP connect
 */
int NetService::post_tcp_connect(user_token *ptoken)
{
    netconnection * pconn = NULL;
    this->connpool.alloc_connection(pconn);
    if(pconn != NULL)
    {
    	/*add lock ..*/
    	pconn->opLock->Lock();
    	{
			pconn->sockfd = socket(AF_INET, SOCK_STREAM, 0);
			pconn->socket_state = S_ACTIVE;
			pconn->OperationType = OP_CONNECT;
			pconn->timeLastAction = time(NULL);
			pconn->token = ptoken;
			if(pconn->sockfd < 0)
			{
				log_fmt(LOG_ERR,"post TCP connect ,new socket occur,code %d",errno);
				pconn->erroCode = CONNECT_ERROR;
				pconn->opLock->Unlock();
				return this->io_close(pconn);
			}
			int reuseaddr = 1;
			setsockopt(pconn->sockfd,SOL_SOCKET,SO_REUSEADDR,&reuseaddr,sizeof(reuseaddr));
			this->setnonblocking(pconn->sockfd);
			if (bind(pconn->sockfd, (struct sockaddr *)&pconn->token->local_addr, sizeof(struct sockaddr)) != 0)
			{
				log_fmt(LOG_ERR,"post_tcp_connect Bind to addr_port occur,code %d",errno);
				pconn->erroCode = CONNECT_ERROR;
				pconn->opLock->Unlock();
				return this->io_close(pconn);
			}
			/*start connect*/
			int ret = connect(pconn->sockfd, (struct sockaddr *) &pconn->token->remote_addr, sizeof(pconn->token->remote_addr));
			if(ret == 0)
			{
				pconn->OperationType = OP_CONNECT_COMPLETE;
				if(this->evtqueue.event_enqueue(EVENT_NEW_CONNECTION,pconn)  != SUCCESS)
				{
						log_fmt(LOG_ERR,"post_tcp_connect event_enqueue  occur error!");
						pconn->erroCode = CONNECT_ERROR;
						pconn->opLock->Unlock();
						return this->io_close(pconn);
				}
				sem_post(&this->io_event_sem);
			}
			else
			{
				if(errno != EINPROGRESS)
				{
					log_fmt(LOG_ERR,"post_tcp_connect,Connect remote host  occur,code %d",errno);
					pconn->erroCode = CONNECT_ERROR;
					pconn->opLock->Unlock();
					return this->io_close(pconn);
				}
				else
				{
					struct epoll_event ev;
					ev.data.fd = pconn->sockfd;
					ev.data.ptr = pconn;
					ev.events = EPOLLOUT | EPOLLET | EPOLLONESHOT;
					if(!pconn->ctr_flag)
					{
						ret = epoll_ctl(this->epfd, EPOLL_CTL_ADD, pconn->sockfd, &ev);
						pconn->ctr_flag = true;
						pconn->events = ev.events;
					}
					else
					{
						ret = epoll_ctl(this->epfd, EPOLL_CTL_MOD, pconn->sockfd, &ev);
						pconn->events = ev.events;
					}

					/*check post flag*/
					if (ret != 0)
					{
						this->log_fmt(LOG_ERR,"post_tcp_connect epoll_ctl error:%d",errno);
						pconn->erroCode = CONNECT_ERROR;
						pconn->opLock->Unlock();
						return this->io_close(pconn);
					}
				}
			}

			/*operation completed*/
			pconn->opLock->Unlock();
    	}
    }

    return SUCCESS;
}
/*
 * POST UDP CONNECT
 */
int NetService::post_udp_connect(user_token *ptoken)
{
    netconnection * pconn = NULL;
    this->connpool.alloc_connection(pconn);
    if(pconn != NULL)
    {
    	/*add lock ..*/
		pconn->opLock->Lock();
		{
			pconn->sockfd = socket(AF_INET, SOCK_DGRAM, 0);
			pconn->socket_state = S_ACTIVE;
			pconn->OperationType = OP_CONNECT;
			pconn->timeLastAction = time(NULL);
			pconn->token = ptoken;
			if(pconn->sockfd < 0)
			{
				log_fmt(LOG_ERR,"post udp connect ,new socket occur,code %d",errno);
				pconn->erroCode = CONNECT_ERROR;
				pconn->opLock->Unlock();
				return this->io_close(pconn);
			}
			int reuseaddr = 1;
			setsockopt(pconn->sockfd,SOL_SOCKET,SO_REUSEADDR,&reuseaddr,sizeof(reuseaddr));
			this->setnonblocking(pconn->sockfd);
			if (bind(pconn->sockfd, (struct sockaddr *)&pconn->token->local_addr, sizeof(struct sockaddr)) != 0)
			{
				log_fmt(LOG_ERR,"post_udp_connect Bind to addr_port occur,code %d",errno);
				pconn->erroCode = CONNECT_ERROR;
				pconn->opLock->Unlock();
				return this->io_close(pconn);
			}

			pconn->OperationType = OP_CONNECT_COMPLETE;
			if(this->evtqueue.event_enqueue(EVENT_NEW_CONNECTION,pconn)   !=  SUCCESS)
			{
					log_fmt(LOG_ERR,"post_udp_connectevent_enqueue occur error!");
					pconn->erroCode = CONNECT_ERROR;
					pconn->opLock->Unlock();
					return this->io_close(pconn);
			}
			sem_post(&this->io_event_sem);

			/*operation completed*/
			pconn->opLock->Unlock();
		}
    }
    return SUCCESS;
}
/*
 * post receive data
 */
int NetService::post_receive(netconnection *pconn)
{
		int ret = -1;
		pconn->opLock->Lock();
		{
				if(pconn->socket_state == S_CLOSE)
				{
					log_fmt(LOG_ERR,"post_receive,error,has closed,connid %u,socket %d,address %x,optype %d ,threadid %u\r\n",pconn->connIndex ,pconn->sockfd,pconn,pconn->OperationType,(unsigned)pthread_self());
					pconn->opLock->Unlock();
					return FAILURE;
				}

				pconn->OperationType = OP_RECV;
				pconn->timeLastAction = time(NULL);
				memset(pconn->iobuf->buffer +  pconn->iobuf->offset,'\0',pconn->iobuf->buf_len - pconn->iobuf->offset);

				struct epoll_event ev;
				ev.data.fd = pconn->sockfd;
				ev.data.ptr = pconn;
				ev.events = EPOLLIN | EPOLLET | EPOLLONESHOT;

				if(!pconn->ctr_flag)
				{
					this->setnonblocking(pconn->sockfd);
					ret = epoll_ctl(this->epfd, EPOLL_CTL_ADD, pconn->sockfd, &ev);
					pconn->ctr_flag = true;
					pconn->events = ev.events;
				}
				else
				{
					ret = epoll_ctl(this->epfd, EPOLL_CTL_MOD, pconn->sockfd, &ev);
					pconn->events = ev.events;
				}
				/*check post flag*/
				if (ret != 0)
				{
					this->log_fmt(LOG_ERR,"post_receive epoll_ctl error:%d",errno);
					pconn->opLock->Unlock();
					return this->io_close(pconn);
				}

				/*post end*/
				pconn->opLock->Unlock();
		}

		/*return result*/
		return ret;
}
/*
 * post send
 */
int NetService::post_send(netconnection *pconn)
{
      int ret = -1;
      pconn->opLock->Lock();
      {
          if(pconn->socket_state == S_CLOSE)
          {
              log_fmt(LOG_ERR,"post_send,error,has closed,connid %u,socket %d,address %x,optype %d ,threadid %u\r\n",pconn->connIndex ,pconn->sockfd,pconn,pconn->OperationType,(unsigned)pthread_self());
              pconn->opLock->Unlock();
              return FAILURE;
          }

          /*go on send*/
          pconn->timeLastAction = time(NULL);
          pconn->OperationType = OP_SEND;

          struct epoll_event ev;
          ev.data.fd = pconn->sockfd;
          ev.data.ptr = pconn;
          ev.events = EPOLLOUT | EPOLLET | EPOLLONESHOT;
          if(!pconn->ctr_flag)
          {
              this->setnonblocking(pconn->sockfd);
              ret = epoll_ctl(this->epfd, EPOLL_CTL_ADD, pconn->sockfd, &ev);
              pconn->ctr_flag = true;
              pconn->events = ev.events;
          }
          else
          {
              ret = epoll_ctl(this->epfd, EPOLL_CTL_MOD, pconn->sockfd, &ev);
              pconn->events = ev.events;
          }
          /*check post flag*/
          if (ret != 0)
          {
              this->log_fmt(LOG_ERR,"post_send epoll_ctl error:%d,connid:%u,ctr_flag:%d",errno,pconn->connIndex,pconn->ctr_flag);
              pconn->opLock->Unlock();
              return this->io_close(pconn);
          }

          /*post end*/
          pconn->opLock->Unlock();
      }

      return ret;
}
/*
 * post trans file
 */
int NetService::post_transfile(netconnection *pconn)
{
    int ret = -1;
    pconn->opLock->Lock();
    {
        if(pconn->socket_state == S_CLOSE)
        {
            log_fmt(LOG_ERR,"post_transfile,error,has closed,connid %u,socket %d,address %x,optype %d ,threadid %u\r\n",pconn->connIndex ,pconn->sockfd,pconn,pconn->OperationType,(unsigned)pthread_self());
            pconn->opLock->Unlock();
            return FAILURE;
        }

        /*trans heartbeat*/
        if(time(NULL) > pconn->timeConnBegin + DEFAULT_TRANSHEARTBEAT_TIME)
        {
        	pconn->timeConnBegin = time(NULL);
        	NEXT_OPRATION op = this->trans_heartbeat(pconn->token);
        	if(op == DO_CLOSE)
        	{
        		 pconn->erroCode = ADMIN_FORCE_CLOSE;
        		 pconn->opLock->Unlock();
        		 return this->io_close(pconn);
        	}
        }

        /*go on transfile*/
        pconn->timeLastAction = time(NULL);
        pconn->OperationType = OP_TRANSFILE;
        this->sock_set_cork(pconn->sockfd);

        struct epoll_event ev;
        ev.data.fd = pconn->sockfd;
        ev.data.ptr = pconn;
        ev.events = EPOLLOUT | EPOLLET | EPOLLONESHOT;
        if(!pconn->ctr_flag)
        {
            this->setnonblocking(pconn->sockfd);
            ret = epoll_ctl(this->epfd, EPOLL_CTL_ADD, pconn->sockfd, &ev);
            pconn->ctr_flag = true;
            pconn->events = ev.events;
        }
        else
        {
            ret = epoll_ctl(this->epfd, EPOLL_CTL_MOD, pconn->sockfd, &ev);
            pconn->events = ev.events;
        }

        /*check post flag*/
        if (ret != 0)
        {
            this->log_fmt(LOG_ERR,"post_transfile epoll_ctl error:%d",errno);

            pconn->opLock->Unlock();
            return this->io_close(pconn);
        }

        /*post end*/
        pconn->opLock->Unlock();
    }

    return ret;
}

/*
 * io accept
 */
int NetService::io_accept()
{
     int acceptfd = 0;
     struct sockaddr_in clientaddr;
     socklen_t clientlen;

     do
     {
         clientlen = sizeof(struct sockaddr);
         acceptfd = accept(this->listenfd,(sockaddr *)&clientaddr, &clientlen);

         if(acceptfd > 0)
         {
        	 this->log_fmt(LOG_ALL,"new acceptfd:%d, freesize:%d, totalsize:%d\r\n",acceptfd,this->connpool.get_free_socket_size() ,this->connpool.get_total_socket_size());

             if(this->connpool.get_total_socket_size()  >=  this->connpool.get_free_socket_size()  + max_connect_num - 10)
             {
                  this->log_fmt(LOG_ERR,"out max connection number!");
                  shutdown(acceptfd,SHUT_RDWR);
                  close(acceptfd);
                  acceptfd = 0;
                  continue;
             }

             /*add to connection list*/
             int ret = 0;
             netconnection * pconn = NULL;
             this->connpool.alloc_connection(pconn);
             if(pconn != NULL)
             {
            	 	pconn->opLock->Lock();
					{
						 pconn->sockfd = acceptfd;
						 pconn->socket_state = S_ACTIVE;
						 pconn->OperationType = OP_ACCEPT_COMPLETE;
						 /*add connection event to event list*/
						 ret = this->evtqueue.event_enqueue(EVENT_NEW_CONNECTION,pconn);
						 /*end lock*/
						 pconn->opLock->Unlock();
					}
					/**
					 * check ret
					 */
					 if(ret == SUCCESS)
					 {
							 sem_post(&this->io_event_sem);
					 }
					 else
					 {
							 this->log_fmt(LOG_ERR,"IO accept  add connection event error!");
							 this->io_close(pconn);
					 }
             }
         }
         else
         {
             if (errno == EAGAIN)// no connect accept
             {
                 break;//end do ... while
             }
             else if (errno == EINTR)
             {
                 //break;//end do ... while
             }
             else
             {
                 if(this->svr_run_flag == 1)
                 {
                     this->log_fmt(LOG_ERR,"connection accept occur error %d !\r\n",errno);

                     /*listen error*/
                     this->re_listen();
                 }
                 break;//end do ... while
             }
         }

     }while(this->svr_run_flag == 1);

     return 0;
}
/*
 * io receive
 */
int NetService::io_recv(netconnection *pconn)
{
    NEXT_OPRATION next_op;
    pconn->opLock->Lock();
    {
        if(pconn->socket_state == S_CLOSE)
        {
            log_fmt(LOG_ERR,"io_recv,error,has closed,connid %u,socket %d,address %x,optype %d ,threadid %u\r\n",pconn->connIndex ,pconn->sockfd,pconn,pconn->OperationType,(unsigned)pthread_self());
            pconn->opLock->Unlock();
            return FAILURE;
        }

        /*
         * read data
         */
        int ret  = -1;
        bool erro_flag = false;
        while(this->svr_run_flag == 1 && pconn->iobuf->buf_len > pconn->iobuf->offset)
        {
            if(pconn->net_protocl == NET_TCP || pconn->net_protocl == NET_RAW_TCP)
            {
                ret = recv(pconn->sockfd, pconn->iobuf->buffer + pconn->iobuf->offset, pconn->iobuf->buf_len - pconn->iobuf->offset,0);
            }
            else
            {
                ret = recvfrom(pconn->sockfd, pconn->iobuf->buffer + pconn->iobuf->offset, pconn->iobuf->buf_len - pconn->iobuf->offset,0,(struct sockaddr *)&pconn->token->remote_addr,&pconn->token->addr_len);
            }

            if (ret > 0)
            {
                pconn->iobuf->offset += ret;
            }
            else if(ret == 0)
            {
                erro_flag = true;
                pconn->erroCode = CLIENT_CLOSED;
                this->log_fmt(LOG_WARN,"io recv zero byte ,error:%d,client connect exit, connid %u,socket %d,address %x,optype %d ,threadid %u!\r\n",errno,pconn->connIndex ,pconn->sockfd,pconn,pconn->OperationType,(unsigned)pthread_self());
                break;
            }
            else if(ret < 0)
            {
                if(errno == EAGAIN || errno == EINTR)
                {
                    break;
                }
                else
                {
                    erro_flag = true;
                    pconn->erroCode = RECV_ERROR;
                    this->log_fmt(LOG_ERR,"io recv,net error:%d,connid %u,socket %d,address %x,optype %d ,threadid %u!\r\n",errno,pconn->connIndex ,pconn->sockfd,pconn,pconn->OperationType,(unsigned)pthread_self());
                    break;
                }
            }
       }

       /*
        * receive compleled notice
        */
       next_op = this->receive_completed(pconn->token,pconn->iobuf);
       pconn->opLock->Unlock();

       /*
        * check is recv error,and close
        */
       if(erro_flag)
       {
           return this->io_close(pconn);
       }
    }

    /*IO completed*/
    return  next_opration(pconn,next_op);
}
/*
 * this function send file
 */
int NetService::io_sendfile(netconnection *pconn)
{
     pconn->opLock->Lock();
     {
         if(pconn->socket_state == S_CLOSE)
         {
             log_fmt(LOG_ERR,"io_sendfile,error,has closed,connid %u,socket %d,address %x,optype %d ,threadid %u\r\n",pconn->connIndex ,pconn->sockfd,pconn,pconn->OperationType,(unsigned)pthread_self());
             pconn->opLock->Unlock();
             return FAILURE;
         }

         /*
          * check is trans completed
          */
         if(pconn->token->filefd < 0 || pconn->token->have_trans_size >= pconn->token->trans_size)
         {
             this->log_fmt(LOG_DEBUG,"trans file complete have send: %lld / %lld ,time:%ld\r\n", pconn->token->have_trans_size,pconn->token->trans_size,(long)time(NULL));
             pconn->opLock->Unlock();
             return this->trans_completed(pconn);
         }

         /*
         *check file offset
         **/
         uint64_t last_len = pconn->token->trans_size - pconn->token->have_trans_size;
         ulong readLen = SENDFILE_SIZE;

         /*
          * send file
          **/
         if(last_len < SENDFILE_SIZE)
         {
            readLen = last_len;
         }
         off64_t offset = (long long)(pconn->token->hFile_offset + pconn->token->have_trans_size);
         ssize_t sendsize = sendfile64(pconn->sockfd,pconn->token->filefd ,&offset,readLen);
         if(sendsize < 0)
         {
             if(errno != EAGAIN)
             {
                 this->log_fmt(LOG_ERR,"IO send file error:%d!\r\n",errno);
                 pconn->erroCode = TRANSMIT_ERROR;
                 pconn->opLock->Unlock();
                 return this->io_close(pconn);
             }
             else
             {
                 sendsize = 0;
                 this->log_fmt(LOG_WARN,"IO send file buf not enough:%d!\r\n",errno);
             }
         }
         /*set current have send size*/
         pconn->token->have_trans_size += sendsize;
         this->log_fmt(LOG_ALL,"io sendfile len:%u, have send: %lld / %lld , file:%lld/%lld,time:%ld\r\n",sendsize , pconn->token->have_trans_size,pconn->token->trans_size,pconn->token->hFile_offset, pconn->token->hFile_size,(long)time(NULL));

         /*go on operation*/
         pconn->opLock->Unlock();
     }

     /*IO next operation*/
     return next_opration(pconn,DO_TRANSFILE);
}
/*
 * io send
 */
int NetService::io_send(netconnection *pconn)
{
    pconn->opLock->Lock();
    {
        if(pconn->socket_state == S_CLOSE)
        {
            log_fmt(LOG_ERR,"io_send,error,has closed,connid %u,socket %d,address %x,optype %d ,threadid %u\r\n",pconn->connIndex ,pconn->sockfd,pconn,pconn->OperationType,(unsigned)pthread_self());
            pconn->opLock->Unlock();
            return FAILURE;
        }

        /*check send buf length,or is send completed*/
        if(pconn->iobuf->size == 0 || pconn->iobuf->offset >= pconn->iobuf->size)
        {
            log_fmt(LOG_DEBUG,"send data completed,connid %u,socket %d,address %x,optype %d ,offset %u,size:%u\r\n",pconn->connIndex ,pconn->sockfd,pconn,pconn->OperationType,pconn->iobuf->offset,pconn->iobuf->size);
            pconn->iobuf->offset = 0;
            /*unlock return*/
            pconn->opLock->Unlock();
            return this->send_completed(pconn);
        }

        /*
         * send buf data
         */
        int ret  = -1;
        while(this->svr_run_flag == 1 && pconn->iobuf->offset < pconn->iobuf->size)
        {
            if(pconn->net_protocl == NET_TCP || pconn->net_protocl == NET_RAW_TCP)
            {
                ret = send(pconn->sockfd, pconn->iobuf->buffer + pconn->iobuf->offset, pconn->iobuf->size - pconn->iobuf->offset,0);
            }
            else
            {
                ret = sendto(pconn->sockfd, pconn->iobuf->buffer + pconn->iobuf->offset, pconn->iobuf->size - pconn->iobuf->offset,0,(struct sockaddr *)&pconn->token->remote_addr,sizeof(struct sockaddr_in));
            }

            if (ret >= 0)
            {
                pconn->iobuf->offset += ret;
            }
            else if(ret < 0)
            {
                if(errno == EAGAIN)
                {
                    this->log_fmt(LOG_WARN,"EAGAIN,send buf not enough,wait ...\r\n");
                    break;
                }
                else
                {
                    this->log_fmt(LOG_ERR,"io send error:%d!\r\n",errno);
                    pconn->erroCode = SEND_ERROR;
                    pconn->opLock->Unlock();
                    return this->io_close(pconn);
                 }
            }
        }

        /*default send completed*/
        pconn->opLock->Unlock();
    }

    /*IO next operation*/
    return next_opration(pconn,DO_SEND);
}
/*
 * io close connect
 */
int NetService::close_connect(netconnection *pconn)
{
    log_fmt(LOG_DEBUG,"close_connect,connid %u,socket %d,address %x,optype %d \r\n",pconn->connIndex ,pconn->sockfd,pconn,pconn->OperationType);
    /*io close event*/
    this->close_completed(pconn->token,pconn->iobuf,pconn->erroCode);

    /*set to close*/
    pconn->socket_state = S_CLOSE;
    pconn->token = NULL;
    /*close socket*/
    if(pconn->sockfd > 0)
    {
        struct epoll_event ev;
        ev.data.fd = pconn->sockfd;
        ev.data.ptr = pconn;
        ev.events = EPOLLIN | EPOLLET;
        epoll_ctl(epfd, EPOLL_CTL_DEL, pconn->sockfd, &ev);
        shutdown(pconn->sockfd,SHUT_RDWR);
        close(pconn->sockfd);
        pconn->sockfd = -1;
    }

    return SUCCESS;
}
/*
 * io close free connect
 */
int NetService::io_close(netconnection *pconn)
{
    pconn->opLock->Lock();
    {
          if(pconn->socket_state == S_CLOSE)
          {
              log_fmt(LOG_ERR,"io_close,error,has closed,connid %u,socket %d,address %x,optype %d ,threadid %u\r\n",pconn->connIndex ,pconn->sockfd,pconn,pconn->OperationType,(unsigned)pthread_self());
              pconn->opLock->Unlock();
              return FAILURE;
          }

          close_connect(pconn);

          pconn->opLock->Unlock();
    }

    if(pconn->list_index == S_WAIT_LIST)
    {
        log_fmt(LOG_ERR,"io_close,check list type error , connid %u,address %x,optype %d",pconn->connIndex ,pconn,pconn->OperationType);
        return FAILURE;
    }

    /*free to pool*/
    return this->connpool.free_connection(pconn);
}


/*
*epool completed function
*/
int  NetService::accept_completed(netconnection *pconn)
{
    NEXT_OPRATION next_op;
    pconn->opLock->Lock();
    {
        if(pconn->socket_state == S_CLOSE)
        {
            pconn->opLock->Unlock();
            return FAILURE;
        }

        next_op = this->accept_completed(pconn->token,pconn->iobuf);
        /*get remote ip*/
        if(pconn->token != NULL)
        {
            pconn->token->conn_index = pconn->connIndex;
            if(SUCCESS != getpeername(pconn->sockfd ,(struct sockaddr *)&pconn->token->remote_addr,&pconn->token->addr_len))
            {
                pconn->token->addr_len = 0;
                this->log_fmt(LOG_WARN,"accept client connect,and getpeername failure !");
            }
        }

        log_fmt(LOG_DEBUG,"accept_completed,connid %u,socket %d,address %x,optype %d \r\n",pconn->connIndex ,pconn->sockfd,pconn,pconn->OperationType);

        pconn->opLock->Unlock();
    }

    return next_opration(pconn,next_op);
}

int  NetService::connect_completed(netconnection *pconn)
{
    NEXT_OPRATION next_op;
    pconn->opLock->Lock();
    {
        if(pconn->socket_state == S_CLOSE)
        {
            pconn->opLock->Unlock();
            return FAILURE;
        }

        pconn->OperationType = OP_CONNECT_COMPLETE;
        next_op = this->connect_completed(pconn->token,pconn->iobuf);

        pconn->opLock->Unlock();
    }

    return    next_opration(pconn,next_op);
}

int  NetService::receive_completed(netconnection *pconn)
{
     NEXT_OPRATION next_op;
     pconn->opLock->Lock();
     {
         if(pconn->socket_state == S_CLOSE)
         {
             pconn->opLock->Unlock();
             return FAILURE;
         }

         pconn->OperationType = OP_RECV_COMPLETE;
         next_op = this->receive_completed(pconn->token,pconn->iobuf);
         pconn->opLock->Unlock();
     }

     return    next_opration(pconn,next_op);
}

int  NetService::send_completed(netconnection *pconn)
{
    NEXT_OPRATION next_op;
    pconn->opLock->Lock();
    {
       if(pconn->socket_state == S_CLOSE)
       {
           pconn->opLock->Unlock();
           return FAILURE;
       }

       pconn->OperationType = OP_SEND_COMPLETE;
       next_op = this->send_completed(pconn->token,pconn->iobuf);

       pconn->opLock->Unlock();
    }

    return  next_opration(pconn,next_op);
}

int  NetService::trans_completed(netconnection *pconn)
{
    NEXT_OPRATION next_op;
    pconn->opLock->Lock();
    {
       if(pconn->socket_state == S_CLOSE)
       {
           pconn->opLock->Unlock();
           return FAILURE;
       }

       pconn->OperationType = OP_TRANSFILE_COMPLETE;
       next_op = this->trans_completed(pconn->token,pconn->iobuf);
       pconn->opLock->Unlock();
    }

    return  next_opration(pconn,next_op);
}

int  NetService::recv_or_send_wait(netconnection *pconn)
{
    NEXT_OPRATION next_op;
    pconn->opLock->Lock();
    {
        if(pconn->socket_state == S_CLOSE)
        {
           pconn->opLock->Unlock();
           return FAILURE;
        }

        next_op = this->recv_or_send_wait_completed(pconn->token,pconn->iobuf);
        pconn->opLock->Unlock();
    }

    return  next_opration(pconn,next_op);
}

/*
 * IO next operation by user define
 **/
int  NetService::next_opration(netconnection *pconn,NEXT_OPRATION next_op)
{
    if(pconn->list_index == S_WAIT_LIST)
    {
        log_fmt(LOG_ERR,"tid %u,next_opration %d,check list type error , connid %u,socket %d,address %x,optype %d \r\n",(unsigned)pthread_self(),next_op,pconn->connIndex ,pconn->sockfd,pconn,pconn->OperationType);
        return FAILURE;
    }

    switch(next_op)
    {
        case  DO_INVALID:
                this->io_close(pconn);
                break;
        case  DO_RECV:
                this->post_receive(pconn);
                break;
        case  DO_SEND:
                this->post_send(pconn);
                break;
        case  DO_TRANSFILE:
                this->post_transfile(pconn);
                break;
        case  DO_CLOSE:
                this->io_close(pconn);
                break;
        case  DO_WAIT:
                this->connpool.switch_to_wait_list(pconn);
                break;
        default:
                this->io_close(pconn);
                break;
    }

    return SUCCESS;
}


/*
*force_close_connection
*/
int NetService::force_close_connection(u_long connIndex)
{
    netconnection *pconn = NULL,* conn = NULL;
    int iconn_listsize = 0;
    bool find_flag = false;
    /*check connlist*/
    connpool.conn_pool_lock.Lock();
    {
        iconn_listsize = connpool.conn_list.size();
        pconn = connpool.conn_list.get_first();

        while(iconn_listsize > 0 && NULL != pconn)
        {
                /*find begin*/
                pconn->opLock->Lock();
                {
                    if(pconn->socket_state != S_CLOSE)
                    {
                        if(pconn->connIndex == connIndex)
                        {
                            find_flag = true;
                            conn = pconn;
                            conn->erroCode = ADMIN_FORCE_CLOSE;
                            this->close_connect(conn);
                        }
                    }

                    /*end*/
                    pconn->opLock->Unlock();
                }
                /*find end*/
                if(find_flag)
                {
                    break;
                }
                /*next node*/
                pconn = pconn->next;
                iconn_listsize--;
        }
        /*check connlist end*/
        connpool.conn_pool_lock.Unlock();
    }

    /*close connect*/
    if(find_flag && conn != NULL)
    {
        log_fmt(LOG_WARN,"force_close_trans_connection, connid %d,op:%d,sock %d",conn->connIndex ,conn->OperationType,conn->sockfd);
        this->connpool.free_connection(conn);
        return SUCCESS;
    }
    else
    {
        return FAILURE;
    }
}

/*
 * io work thread
 */
void * NetService::io_working_proces(void *pParam)
{
		NetService * pSvr = (NetService *)pParam;
		if(pSvr == NULL)
				return NULL;

		struct epoll_event events[MAXEVENTS];
		int i = 0;
		int nfds = 0;
		while(pSvr->svr_run_flag == 1)
		{
			nfds = epoll_wait(pSvr->epfd, events, MAXEVENTS,-1);

			for(i = 0; i < nfds && pSvr->svr_run_flag == 1 ; i++)
			{
					/*
					 * check fd is null
					 */
					if(events[i].data.fd  <= 0 )
					{
						continue;
					}

					/*
					 * check if new connection come in.
					 */
					if (events[i].data.fd == pSvr->listenfd)
					{
							if (events[i].events & EPOLLIN)
							{
									sem_post(&pSvr->io_control_sem);
							}
							else if ((events[i].events & EPOLLERR) || (events[i].events & EPOLLHUP))
							{
								  /*
								   * listen socket error,and must close ,then recreate listen
								   */
								  pSvr->re_listen();
								  break;//end for
							}
							else
							{
									pSvr->re_listen();
									break;//end for
							}
					}
					else /*handle user connection*/
					{
							netconnection * pconn = NULL;
							pconn = (netconnection *) events[i].data.ptr;
							if(pconn == NULL)
							{
									 pSvr->log_fmt(LOG_ERR,"handle user connection error,get netconnection ptr is null,fd:%d",events[i].data.fd);
									 close(events[i].data.fd);
									 continue;
							}

							/*add event to event list*/
							pconn->opLock->Lock();
							{
									if(pconn->list_index != S_CONN_LIST)
									{
										   pSvr->log_fmt(LOG_WARN,"handle user connection warn,connection status:%d,fd:%d",pconn->list_index,events[i].data.fd);
										   pconn->opLock->Unlock();
										   continue;
									}

									/*
									 * check event type
									 */
									if (events[i].events & EPOLLIN)
									{
											if(pconn->OperationType == OP_RECV)
											{
													pconn->OperationType = OP_RECV_COMPLETE;
													pSvr->evtqueue.event_enqueue(EVENT_RECV,pconn);
													sem_post(&pSvr->io_event_sem);
											}
											else
											{
													pSvr->log_fmt(LOG_WARN,"last operation receive not handler completed!");
											}
									}
									else  if (events[i].events & EPOLLOUT)
									{
											if(pconn->OperationType == OP_SEND)
											{
												   pconn->OperationType = OP_SEND_COMPLETE;
												   pSvr->evtqueue.event_enqueue(EVENT_SEND,pconn);
												   sem_post(&pSvr->io_event_sem);
											}
											else if(pconn->OperationType == OP_TRANSFILE)
											{
												   pSvr->io_queue.event_enqueue(EVENT_TRANS,pconn);
												   sem_post(&pSvr->io_send_sem);
											}
											else if(pconn->OperationType == OP_CONNECT)
											{
												   pconn->OperationType = OP_CONNECT_COMPLETE;
												   pSvr->evtqueue.event_enqueue(EVENT_NEW_CONNECTION,pconn);
												   sem_post(&pSvr->io_event_sem);
											}
											else
											{
													pSvr->log_fmt(LOG_WARN,"last operation EPOLLOUT not handler completed!");
											}
									}
									else  if (events[i].events & EPOLLERR)
									{
											pconn->erroCode = SERVER_ERROR;
											pSvr->evtqueue.event_enqueue(EVENT_CLIENT_ERROR,pconn);
											sem_post(&pSvr->io_event_sem);
									}
									else  if (events[i].events & EPOLLHUP)
									{
											pconn->erroCode = CLIENT_CLOSED;
											pSvr->evtqueue.event_enqueue(EVENT_CLIENT_CLOSED,pconn);
											sem_post(&pSvr->io_event_sem);
									}
									else
									{
											pconn->erroCode = CLIENT_ERROR;
											pSvr->evtqueue.event_enqueue(EVENT_CLIENT_ERROR,pconn);
											sem_post(&pSvr->io_event_sem);
									}

									/*handle end*/
									pconn->opLock->Unlock();
							}
					}
				}
		}

		/*thread exit*/
		pSvr->log_fmt(LOG_WARN,"work thread exit:%u\r\n",(unsigned)pthread_self());
		return pSvr;
}

/*
 * io control thread
 */
void * NetService::io_control_proces(void *pParam)
{
     NetService * pSvr = (NetService *)pParam;
     if(pSvr == NULL)
         return NULL;

     while(pSvr->svr_run_flag == 1)
     {
         /*
          * wait io event
          */
         sem_wait(&pSvr->io_control_sem);

         pSvr->io_accept();
     }

     /*thread exit*/
     pSvr->log_fmt(LOG_WARN,"control thread exit:%u\r\n",(unsigned)pthread_self());
     return pSvr;
}

/*
 * io_event_proces thread
 */
void * NetService::io_event_proces(void *pParam)
{
     NetService * pSvr = (NetService *)pParam;
     if(pSvr == NULL)
         return NULL;

     int  dwError;
     EVENT ioEvent;
     netconnection *pconn = NULL;

     while(pSvr->svr_run_flag == 1)
     {
         /*
          * wait io event
          */
         sem_wait(&pSvr->io_event_sem);

         /*
          * get io event
          **/
         pconn = NULL;
         dwError = pSvr->evtqueue.event_dequeue(&ioEvent,(void **)&pconn);

         if(pSvr->evtqueue.get_event_count() > 0)
         {
             sem_post(&pSvr->io_event_sem);
         }
         if(dwError != SUCCESS || pconn->socket_state == S_CLOSE)
         {
              continue;
         }
         else
         {
             pconn->CurrentEvent = ioEvent;
         }

         /*handle io event*/
         switch(ioEvent)
         {
             case EVENT_NEW_CONNECTION:
                 {
                     if(pconn->OperationType == OP_CONNECT_COMPLETE)
                     {
                         pSvr->connect_completed(pconn);
                     }
                     else if(pconn->OperationType == OP_ACCEPT_COMPLETE)
                     {
                         pSvr->accept_completed(pconn);
                     }
                     else
                     {
                         pSvr->io_close(pconn);
                     }
                 }
                 break;
             case EVENT_RECV:
                 pSvr->io_recv(pconn);
                 break;
             case EVENT_RECV_COMPLETE:
                 pSvr->receive_completed(pconn);
                 break;
             case EVENT_SEND:
                 pSvr->io_send(pconn);
                 break;
             case EVENT_SEND_COMPLETE:
                 pSvr->send_completed(pconn);
                 break;
             case EVENT_TRANS_COMPLETE:
                 pSvr->trans_completed(pconn);
                 break;
             case EVENT_CLIENT_ERROR:
                 pSvr->io_close(pconn);
                 break;
             case EVENT_CLIENT_CLOSED:
                 pSvr->io_close(pconn);
                 break;
             default:break;
         }
     }

     /*thread exit*/
     pSvr->log_fmt(LOG_WARN,"event thread exit:%u\r\n",(unsigned)pthread_self());
     return pSvr;
}

/*
 * io_clean_proces thread
 */
void * NetService::io_clean_proces(void *pParam)
{
    NetService * pSvr = (NetService *)pParam;
    if(pSvr == NULL)
        return NULL;

    /*time out handel*/
    netconnection * pconn = NULL;
    int iconn_listsize = 0;
    int time_out = 0;

    /*wait handel*/
    netconnection * temp_pconn = NULL;
    stack<netconnection *> timout_list;
    time_t timeCur = time(NULL);
    BOOL isTimeOut = FALSE;

    while(pSvr->svr_run_flag == 1)
    {
        pSvr->connpool.conn_pool_lock.Lock();
        {
            pconn = NULL;
            iconn_listsize = pSvr->connpool.conn_list.size();
            pconn = pSvr->connpool.conn_list.get_first();
            timeCur = time(NULL);

            while(iconn_listsize > 0 && NULL != pconn)
            {
                pconn->opLock->Lock();
                {
                    if(pconn->socket_state != S_CLOSE)
                    {
                        isTimeOut = FALSE;
                        /*default system time out*/
                        time_out = DEFAULT_TIMEOUT;
                        /*check opration time out*/
                        if(pconn->OperationType == OP_RECV)
                        {
                            if( pconn->token != NULL &&  pconn->token->receive_timeout >0)
                            {
                                  time_out = pconn->token->receive_timeout;
                            }
                            if(timeCur - pconn->timeLastAction >= time_out)
                            {
                                  isTimeOut = TRUE;
                            }
                        }
                        else if(pconn->OperationType == OP_SEND)
                        {
                            if( pconn->token != NULL &&  pconn->token->send_timeout >0)
                            {
                                time_out = pconn->token->send_timeout;
                            }
                            if(timeCur - pconn->timeLastAction >= time_out)
                            {
                                isTimeOut = TRUE;
                            }
                        }
                        else if(pconn->OperationType == OP_TRANSFILE)
                        {
                             if( pconn->token != NULL &&  pconn->token->trans_file_timeout >0)
                             {
                                 time_out = pconn->token->trans_file_timeout;
                             }
                             if(timeCur - pconn->timeLastAction >= time_out)
                             {
                                 isTimeOut = TRUE;
                             }
                        }

                        /*check is time out and close*/
                        if(isTimeOut == TRUE)
                        {
								pSvr->log_fmt(LOG_WARN,"OP time out connid %d,op:%d,sock %d",pconn->connIndex ,pconn->OperationType,pconn->sockfd);
								pSvr->close_connect(pconn);
								timout_list.push(pconn);
                        }
                    }

                    /*end*/
                    pconn->opLock->Unlock();
                }

                /*next node*/
                pconn = pconn->next;
                iconn_listsize--;
            }
            /*check connlist end*/
            pSvr->connpool.conn_pool_lock.Unlock();
        }


        /*
         * sleep THREAD_TIME_TICK
         * and next run
         */
        usleep(THREAD_TIME_TICK);

        /*
        *handle timeout connection
        */
        while(timout_list.size() > 0)
        {
            	temp_pconn = timout_list.top();
            	timout_list.pop();
            	pSvr->connpool.free_connection(temp_pconn);
        }
    }

    /*thread exit*/
    pSvr->log_fmt(LOG_WARN,"clean thread exit:%u\r\n",(unsigned)pthread_self());
    return pSvr;
}

/*
 * net_opration_proces thread
 */
void * NetService::net_opration_proces(void *pParam)
{
    NetService * pSvr = (NetService *)pParam;
    if(pSvr == NULL)
        return NULL;

    /*time tick*/
    int time_tick = 0;
    /*net opration*/
    NETPROTOCL net_protocl;
    user_token * token = NULL;
    /*wait opration*/
    netconnection * pconn = NULL;
    int wait_listsize = 0;
    stack<netconnection *> wait_list;
    stack<NEXT_OPRATION> wait_op_list;
    NEXT_OPRATION net_op;
    /*max limit hand count :100*/
    int hand_count = 100;
    while(pSvr->svr_run_flag == 1)
    {
        /*net opration*/
        if(time_tick == 0)
        {
            /*hand limit 100*/
            hand_count = 100;
            /*hand*/
            while(pSvr->net_op_queue.get_opration_count() > 0 && hand_count > 0)
            {
                if(pSvr->net_op_queue.op_dequeue(&net_protocl,&token) == SUCCESS)
                {
                    switch(net_protocl)
                    {
                        case NET_TCP:
                            pSvr->post_tcp_connect(token);
                            break;
                        case NET_UDP:
                            pSvr->post_udp_connect(token);
                            break;
                        default:break;
                    }
                }
                else
                {
                    break;
                }

                /*hand count*/
                hand_count --;
            }
            /*time tick reset to one*/
            time_tick = 1;
        }
        else if(time_tick >= 1) /*handle wait list opration*/
        {
            pSvr->connpool.wait_pool_lock.Lock();
            {
                pconn = NULL;
                wait_listsize = pSvr->connpool.wait_list.size();
                pconn = pSvr->connpool.wait_list.get_first();
                while(wait_listsize > 0 && NULL != pconn)
                {
                      net_op = pSvr->recv_or_send_wait_completed(pconn->token,pconn->iobuf);

                      if(net_op != DO_WAIT)
                      {
                          wait_op_list.push(net_op);
                          wait_list.push(pconn);
                      }

                      pconn = pconn->next;
                }

                pSvr->connpool.wait_pool_lock.Unlock();
            }
            /*handle wait opration*/
            while(wait_list.size() > 0 && wait_op_list.size() > 0)
            {
                pconn = wait_list.top();
                wait_list.pop();

                net_op = wait_op_list.top();
                wait_op_list.pop();

                pSvr->connpool.switch_to_conn_list(pconn);
                pSvr->next_opration(pconn,net_op);
            }
            /*time tick reset to zero*/
            time_tick = 0;
        }

        /*wait for 0.01 seconds*/
         usleep(10000);
    }

    /*thread exit*/
    pSvr->log_fmt(LOG_WARN,"net opration thread exit:%u\r\n",(unsigned)pthread_self());
    return pSvr;
}

/*
 * io send file thread
 */
void * NetService::io_sendfile_proces(void *pParam)
{
     NetService * pSvr = (NetService *)pParam;
     if(pSvr == NULL)
         return NULL;

     int  dwError;
     EVENT ioEvent;
     netconnection *pconn = NULL;
     /*
     *thread run
     */
     while(pSvr->svr_run_flag == 1)
     {
         /*
          * wait send  task
          */
         sem_wait(&pSvr->io_send_sem);

         /*
         * get io event
         **/
         pconn = NULL;
         dwError = pSvr->io_queue.event_dequeue(&ioEvent,(void **)&pconn);
         if(pSvr->io_queue.get_event_count() > 0)
         {
             sem_post(&pSvr->io_send_sem);
         }
         if(dwError != SUCCESS || pconn->socket_state == S_CLOSE)
         {
             continue;
         }
         /*send file*/
         if(ioEvent == EVENT_TRANS)
         {
             pSvr->io_sendfile(pconn);
         }
         else
         {
             pSvr->io_close(pconn);
         }
     }

     /*thread exit*/
     pSvr->log_fmt(LOG_WARN,"io_sendfile_proces thread exit:%u\r\n",(unsigned)pthread_self());
     return pSvr;
}
